package io.gitee.thinkbungee.utils.mongo;

import com.mongodb.BasicDBObject;
import com.mongodb.MongoCommandException;
import com.mongodb.client.AggregateIterable;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoCursor;
import io.gitee.thinkbungee.entity.mongo.MongoDefinition;
import org.bson.Document;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.RecursiveTask;
import java.util.stream.Collectors;

/**
 * @author chenyichang
 * @date 2022/3/9
 */
public class MongoScanner {

    private final Logger logger = LoggerFactory.getLogger(getClass());

    /**
     * mongo的链接
     */
    private MongoCollection<Document> collection;

    /**
     * 最多扫描10万条数据
     */
    private final static int DEFAULT_SCAN_COUNT = 100000;

    /**
     * 最终集合的所有列名
     */
    private List<String> colNames;

    /**
     * 表的所有数据
     */
    private MongoDefinition mongoDefinition;

    public MongoScanner(MongoCollection<Document> collection) {
        this.collection = collection;
        scan();
    }

    /**
     * 扫描表
     */
    private void scan() {
        //初始化并解析
        initColNames();
        mongoDefinition = scanType();
        MongoManager.putInfo(collection.getNamespace().getCollectionName(), mongoDefinition);
        //释放资源
        this.collection = null;
    }

    private MongoDefinition scanType() {
        MongoDefinition result = new MongoDefinition();
        ForkJoinPool pool = new ForkJoinPool();
        ForkJoinProcessType task = new ForkJoinProcessType(colNames);
        List<MongoDefinition> invoke = pool.invoke(task);
        result.setChild(invoke);
        result.setPropertyName(collection.getNamespace().getCollectionName());
        return result;
    }

    /**
     * 解析集合的列名
     * 用forkJoin框架实现
     */
    private void initColNames() {
        long start = System.currentTimeMillis();
        //最大扫描10万条
        int scan = DEFAULT_SCAN_COUNT;
        //总条数
        long count = this.collection.countDocuments();

        //使用ForkJoin框架进行扫描
        ForkJoinPool pool = new ForkJoinPool();

        ForkJoinTask<List<String>> task;

        //最多扫描10万条数据
        if (count > scan) {
            task = new ForkJoinGetProcessName(0, scan);
        } else {
            task = new ForkJoinGetProcessName(0, (int) count);
        }

        this.colNames = pool.invoke(task);

        logger.info(String.format("collection[%s]初始化列名成功，用时%s毫秒",
                collection.getNamespace().getCollectionName(),
                System.currentTimeMillis() - start));

    }

    public MongoDefinition getProduct() {
        return mongoDefinition;
    }

    /**
     * 解析属性类型
     * 参考文档：https://www.runoob.com/mongodb/mongodb-operators-type.html
     *
     * @param propertyName 属性名
     * @return
     */
    private MongoDefinition processNameType(String propertyName) {
        //4：数组
        int arrayType = 4;
        /**
         * 1：double
         * 2：string
         * 3：对象
         * 4：数组
         * 8：boolean
         * 9：date
         * 16：int
         * 18：long
         * @see MongoType
         */
        int[] types = {3, 16, 18, 8, 9, 2, 1};
        //结果
        MongoDefinition result = new MongoDefinition();

        //主键id
        if ("_id".equals(propertyName)) {
            result.setType(MongoType.OBJECT_ID.getNum());
            result.setPropertyName("_id");
            return result;
        }
        result.setPropertyName(propertyName);
        MongoCursor<Document> isArray = collection.find(new Document(propertyName, new Document("$type", arrayType))).limit(1).iterator();
        if (isArray.hasNext()) {
            result.setArray(true);
            for (int type : types) {
                MongoCursor<Document> iter = collection.find(new Document(propertyName, new Document("$type", type))).limit(1).iterator();
                if (iter.hasNext()) {
                    //如果是对象，处理该对象
                    if (type == 3) {
                        result.setChild(this.produceChildList(propertyName));
                    }
                    result.setType(type);
                    logger.info(String.format("解析[%s]类型是%s", propertyName, MongoType.typeInfo(type)));
                    return result;
                }
            }
        } else {
            for (int type : types) {
                MongoCursor<Document> iter = collection.find(new Document(propertyName, new Document("$type", type))).limit(1).iterator();
                if (iter.hasNext()) {
                    //如果是对象，处理该对象
                    if (type == 3) {
                        result.setChild(this.produceChildList(propertyName));
                    }
                    result.setType(type);
                    logger.info(String.format("解析[%s]类型是%s", propertyName, MongoType.typeInfo(type)));
                    return result;
                }
            }
            result.setType(2);
        }
        logger.info(String.format("解析[%s]类型是%s", propertyName, MongoType.typeInfo(result.getType())));
        return result;
    }

    private List<MongoDefinition> produceChildList(String parentName) {
        Set<String> childNamesSet = this.getNextParameterNames(parentName);
        ArrayList<String> strings = new ArrayList<>(childNamesSet);
        List<String> childNames = strings.stream().map(name -> parentName + "." + name).collect(Collectors.toList());
        ForkJoinPool pool = new ForkJoinPool();
        ForkJoinProcessType task = new ForkJoinProcessType(childNames);
        return pool.invoke(task);
    }

    /**
     * 解析下一级对象名
     * 如:  {"user":{"name":"张三",age:12}}
     * 传入user--->返回 [name,age]
     *
     * @param parentName 参数名，该参数必须是对象类型
     * @return
     */
    private Set<String> getNextParameterNames(String parentName) {
        String unwindName = parentName;
        if (parentName.contains(".")) {
            unwindName = parentName.split("\\.")[0];
        }
        AggregateIterable<Document> aggregate = collection.aggregate(Arrays.asList(
                new Document("$match", new Document(parentName, new Document("$exists", true))),
                new Document("$unwind", "$" + unwindName),
                new Document("$limit", 3000),
                new Document("$project", new Document("list", "$" + parentName)
                        .append("_id", false)),
                new Document("$unwind", "$list")
        ));
        Set<String> namesSet = new HashSet<>();
        for (Document document : aggregate) {
            Object list = document.get("list");
            if (list instanceof Map) {
                Set<String> names = ((Document) list).keySet();
                namesSet.addAll(names);
            }
        }
        logger.info(String.format("解析%s有%s个子属性", parentName, namesSet.size()));
        return namesSet;
    }


    /**
     * 实现
     */
    class ForkJoinGetProcessName extends RecursiveTask<List<String>> {
        /**
         * 开始游标
         */
        private final int begin;
        /**
         * 结束游标
         */
        private final int end;

        public ForkJoinGetProcessName(int begin, int end) {
            this.begin = begin;
            this.end = end;
        }

        @Override
        protected List<String> compute() {
            /**
             * 一次任务：最多5000条数据
             */
            int threshold = 5000;

            int count = end - begin;

            if (threshold >= count) {
                //执行任务
                return groupAggregation(begin, count);
            } else {
                //拆分任务
                int middle = (begin + end) / 2;
                ForkJoinGetProcessName pre = new ForkJoinGetProcessName(begin, middle);
                pre.fork();
                ForkJoinGetProcessName next = new ForkJoinGetProcessName(middle + 1, end);
                next.fork();
                return distinctAndJoin(pre.join(), next.join());
            }
        }

        /**
         * 去重并合并
         */
        private List<String> distinctAndJoin(List<String> a, List<String> b) {
            a.removeAll(b);
            a.addAll(b);
            return a;
        }

        /**
         * 分组 聚合
         * 可以查看doc/knowledge/mongo_aggregation.txt
         */
        public List<String> groupAggregation(Integer skip, Integer limit) throws MongoCommandException {
            skip = skip == null ? 0 : skip;
            limit = limit == null ? DEFAULT_SCAN_COUNT : limit;

            List<BasicDBObject> dbObjects = Arrays.asList(
                    //将每一个文档都拆分为
                    // k : key
                    // v : value
                    new BasicDBObject("$project",
                            new BasicDBObject("arrayofkeyvalue",
                                    new BasicDBObject("$objectToArray", "$$ROOT"))),
                    new BasicDBObject("$skip", skip),//跳过已经计算过的任务
                    new BasicDBObject("$limit", limit),//有多少条就计算多少条
                    //将上面的结果数组进行拆分
                    new BasicDBObject("$unwind", "$arrayofkeyvalue"),
                    //分组，并把k的值放进数组
                    new BasicDBObject("$group",
                            new BasicDBObject("_id", "null").append("allkeys",
                                    new BasicDBObject("$addToSet", "$arrayofkeyvalue.k")))
            );
            AggregateIterable<Document> aggregate = collection.aggregate(dbObjects);
            Document document = aggregate.first();
            if (document == null) {
                MongoCursor<Document> existsList = collection.find(
                        new BasicDBObject("$ROOT", new BasicDBObject("$exists", true))
                ).limit(100).iterator();
                Set<String> keyset = new HashSet<>();
                while (existsList.hasNext()) {
                    Document next = existsList.next();
                    Map<String, Object> keyMap = (Document) next.get("$ROOT");
                    keyset.addAll(keyMap.keySet());
                }
                return new ArrayList<>(keyset);
            } else {
                return (List<String>) document.get("allkeys");
            }


        }
    }

    /**
     * 解析类型
     */
    class ForkJoinProcessType extends RecursiveTask<List<MongoDefinition>> {
        List<String> names;
        private final int THRESHOLD = 6;

        public ForkJoinProcessType(List<String> names) {
            this.names = names;
        }

        @Override
        protected List<MongoDefinition> compute() {
            if (names.size() <= THRESHOLD) {
                List<MongoDefinition> result = new ArrayList<>();
                for (String name : names) {
                    //解析 该name的 类型
                    MongoDefinition md = processNameType(name);
                    result.add(md);
                }
                return result;
            } else {
                int size = names.size();
                int middle = size / 2;
                List<String> left = names.subList(0, middle);
                List<String> right = names.subList(middle, size);
                ForkJoinProcessType pre = new ForkJoinProcessType(left);
                pre.fork();
                ForkJoinProcessType next = new ForkJoinProcessType(right);
                next.fork();
                return mergeList(pre.join(), next.join());
            }

        }

        /**
         * 合并
         */
        private <T> List<T> mergeList(List<T> join1, List<T> join2) {
            join1.addAll(join2);
            return join1;
        }
    }

}
